Flask 源码阅读：下篇

这是上两篇：

flask项目大名鼎鼎，不需要多做介绍。我把它称之为python服务开发的TOP2项目，另外一个就是django。这两个项目各有千秋，各自有不同的应用场景，都需要深入理解，熟练掌握。本次源码选择的版本是 1.1.2，我会采用慢读法，尽自己最大努力把它讲透。本篇是第3篇-自助餐，主要包括:

• view 解析
• blueprint 解析
• 小结

view 解析

flask一个简单的监听函数如下:

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        return do_the_login()
    else:
        return show_the_login_form()

如果URL多了，就需要实现多个监听函数，代码会比较乱。同时对于一个URL，在监听函数中区分http的method，进行不同的业务逻辑处理。一个函数中处理两种逻辑，也不太符合单一职责，会让代码难以维护。

这种情况下，就需要使用视图。下面是一个视图示例:

class CounterAPI(MethodView):

    def get(self):
        return session.get('counter', 0)

    def post(self):
        session['counter'] = session.get('counter', 0) + 1
        return 'OK'

app.add_url_rule('/counter', view_func=CounterAPI.as_view('counter'))

CounterAPI可以把一个类的实例方法注册到一个URL上，自动将get和post方法区分开。我们一起看看View的实现，先是所有View的基类:

class View(object):
    
    @classmethod
    def as_view(cls, name, *class_args, **class_kwargs):
        
        def view(*args, **kwargs):
            self = view.view_class(*class_args, **class_kwargs)
            return self.dispatch_request(*args, **kwargs)
        ...
        view.view_class = cls
        view.__name__ = name
        ...
        return view

as_view函数返回一个视图函数，在视图函数里可以派发和处理request。

MethodViewType是一个元类，定义了视图支持的所有HTTP方法的集合:

http_method_funcs = frozenset(
    ["get", "post", "head", "options", "delete", "put", "trace", "patch"]
)

class MethodViewType(type):

    def __init__(cls, name, bases, d):
        super(MethodViewType, cls).__init__(name, bases, d)

        if "methods" not in d:
            methods = set()
            ...
            for key in http_method_funcs:
                if hasattr(cls, key):
                    methods.add(key.upper())

            if methods:
                cls.methods = methods

MethodView是使用MethodViewType和View创建的新类：

class MethodView(with_metaclass(MethodViewType, View)):

    def dispatch_request(self, *args, **kwargs):
        meth = getattr(self, request.method.lower(), None)
        ...
        return meth(*args, **kwargs)

with_metaclass 是为了兼容python2的语法，可以简单的理解为继承自MethodViewType和View

dispatch_request中根据请求的http-method找到对应的方法，进行执行。

view的处理函数还可以增加装饰器，示例如下:

# 使用示例
class SecretView(View):
    methods = ['GET']
    decorators = [login_required]
            
class View(object):
    
    @classmethod
    def as_view(cls, name, *class_args, **class_kwargs):
        ...
        if cls.decorators:
            view.__name__ = name
            view.__module__ = cls.__module__
            # 包装上装饰器
            for decorator in cls.decorators:
                view = decorator(view)
        ...
        return view

# 装饰器
def login_required(view):
    @functools.wraps(view)
    def wrapped_view(**kwargs):
        if g.user is None:
            return redirect(url_for('auth.login'))

        return view(**kwargs)

    return wrapped_view

blueprint 解析

View相对还是比较单薄，大型的项目都会分模块进行开发，所以flask还有blueprint的概念。下面是示例项目flaskr中的 auth.py :

import functools

from flask import (
    Blueprint, flash, g, redirect, render_template, request, session, url_for
)

...

bp = Blueprint('auth', __name__, url_prefix='/auth')

@bp.route('/register', methods=('GET', 'POST'))
def register():
    ...

@bp.route('/login', methods=('GET', 'POST'))
def login():
    ...
    
@bp.route('/logout')
def logout():
    ...

这里定义了一个名称叫做auth的蓝图，里面定义了3个方法: register , login 和 logout 。蓝图在app的__init__.py中注册:

def create_app():
    app = ...
    # existing code omitted

    from . import auth
    app.register_blueprint(auth.bp)

    return app

在flask项目中还有名为blog的蓝图，提供博客文章的增删改查方法:

bp = Blueprint("blog", __name__)


@bp.route("/")
def index():
    ...
    
@bp.route("/create", methods=("GET", "POST"))
@login_required
def create():
    ...

采用这种方式，就可以很方便的分模块进行程序开发。

了解了bluerpint的使用方法后，我们再看看其实现原理。

class Blueprint(_PackageBoundObject):
    
    def __init__(
        self,
        name,
        import_name,
        static_folder=None,
        static_url_path=None,
        template_folder=None,
        url_prefix=None,
        subdomain=None,
        url_defaults=None,
        root_path=None,
        cli_group=_sentinel,
    ):
        _PackageBoundObject.__init__(
            self, import_name, template_folder, root_path=root_path
        )
        self.name = name
        self.url_prefix = url_prefix
        self.subdomain = subdomain
        self.static_folder = static_folder
        self.static_url_path = static_url_path
        self.deferred_functions = []
        if url_defaults is None:
            url_defaults = {}
        self.url_values_defaults = url_defaults
        self.cli_group = cli_group

上面Blueprint的构造函数中显示：

• 继承自_PackageBoundObject。_PackageBoundObject上一篇介绍过，主要实现本地目录的动态加载，因为蓝图也有一些模版需求，所以继承了_PackageBoundObject。
• deferred_functions数组是蓝图的所有视图的集合
• url_prefix，subdomain， static_folder等是蓝图模块化的功能参数

蓝图的route装饰器:

def route(self, rule, **options):
    def decorator(f):
        endpoint = options.pop("endpoint", f.__name__)
        self.add_url_rule(rule, endpoint, f, **options)
        return f

    return decorator

def add_url_rule(self, rule, endpoint=None, view_func=None, **options):
    ...
    self.record(lambda s: s.add_url_rule(rule, endpoint, view_func, **options))

def record(self, func):
    self.deferred_functions.append(func)

这里主要的疑问在添加视图函数时候的lambda函数的参数 s 是什么？继续看看蓝图的注册:

# app的方法
def register_blueprint(self, blueprint, **options):

    self.blueprints[blueprint.name] = blueprint
    self._blueprint_order.append(blueprint)
    first_registration = True

    blueprint.register(self, options, first_registration)

# blueprint的方法
def register(self, app, options, first_registration=False):
    self._got_registered_once = True
    state = self.make_setup_state(app, options, first_registration)

    for deferred in self.deferred_functions:
        deferred(state)
   ...

make_setup_stat创建BlueprintSetupState对象, 然后执行蓝图route添加到deferred_functions的方法。这个方法就是前面的lambda函数，前面的 s 就是state对象.

class BlueprintSetupState(object):
    
    def __init__(self, blueprint, app, options, first_registration):
        #: a reference to the current application
        self.app = app
        self.blueprint = blueprint
    
    def add_url_rule(self, rule, endpoint=None, view_func=None, **options):
        """A helper method to register a rule (and optionally a view function)
        to the application.  The endpoint is automatically prefixed with the
        blueprint's name.
        """
        if self.url_prefix is not None:
            if rule:
                rule = "/".join((self.url_prefix.rstrip("/"), rule.lstrip("/")))
            else:
                rule = self.url_prefix
        options.setdefault("subdomain", self.subdomain)
        if endpoint is None:
            endpoint = _endpoint_from_view_func(view_func)
        defaults = self.url_defaults
        if "defaults" in options:
            defaults = dict(defaults, **options.pop("defaults"))
        self.app.add_url_rule(
            rule,
            "%s.%s" % (self.blueprint.name, endpoint),
            view_func,
            defaults=defaults,
            **options
        )

BlueprintSetupState中建立了app和blueprint的关联，并且使用app的add_url_rule方法，把blueprint的视图函数注册进入app。

小结

flask是一个 micro 框架，但是也(至少)可以支持中型项目。我们可以利用Blueprint和View功能进行模块化: View可以很好的区分URL上的http-method；Blueprint可以很好的定义子域名和URL前缀等。

ps:本文为该系列文章第三篇，关注【Python开发者】公众号，锁定更多精彩内容。

- EOF -

推荐阅读  点击标题可跳转

觉得本文对你有帮助？请分享给更多人

推荐关注「Python开发者」，提升Python技能

点赞和在看就是最大的支持❤️